package com.breeze.components.message.delay.executor;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.breeze.components.core.util.NormalRun;
import com.breeze.components.message.delay.DelayQueueProperties;
import com.breeze.components.message.delay.holder.DelayQueueHolder;
import com.breeze.components.message.delay.msg.DelayMessage;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;

/**
 * 延迟队列执行器工厂
 *
 * @author breeze
 */
@Slf4j
public class DelayQueueExecutorEngine<T extends DelayMessage> extends NormalRun {

  private final DelayQueueHolder delayQueueHolder;

  /** 队列 */
  private final String queue;

  /** 执行器 */
  private final DelayQueueExecutor<T> delayQueueExecutor;

  /** 数据类型 */
  private final Class<T> classType;

  /** 失败重试次数 */
  private Integer tryTimes = DelayQueueProperties.getMaxRetryTime();

  /** 失败重试时间间隔 */
  private Long retryDelay = 1000L;

  public DelayQueueExecutorEngine(
      DelayQueueHolder redisDelayQueueHolder,
      DelayQueueExecutor<T> delayQueueExecutor,
      Class<T> classType) {
    log.info("{}延时任务引擎初始化", delayQueueExecutor.queue());
    this.delayQueueHolder = redisDelayQueueHolder;
    this.delayQueueExecutor = delayQueueExecutor;
    this.queue = delayQueueExecutor.queue();
    this.classType = classType;
    // 初始化
    redisDelayQueueHolder.initDelayQueue(queue);
    // 启动线程
    this.startExecutor();
  }

  public DelayQueueExecutorEngine<T> tryTimes(int tryTimes) {
    if (tryTimes > 0) {
      this.tryTimes = tryTimes;
    }
    return this;
  }

  public DelayQueueExecutorEngine<T> retryDelay(Long reTryDelay) {
    if (reTryDelay > 0) {
      this.retryDelay = reTryDelay;
    }
    return this;
  }

  private void error(T message) {
    try {
      int errorTimes = message.getErrorTimes();
      if (errorTimes < tryTimes) {
        log.info("延迟队列-{}重试, message:{}, errorTimes:{}", queue, message, errorTimes);
        message.setErrorTimes(errorTimes + 1);
        delayQueueHolder.addJob(message, retryDelay, TimeUnit.MILLISECONDS, queue);
      }
    } catch (Exception e) {
      log.error("延迟队列-{}重试失败,请核对, message:{}", queue, message, e);
    }
  }

  private void startExecutor() {
    Thread thread =
        new Thread(
            () -> {
              try {
                log.info("启动延迟队列-{}监听线程", queue);
                while (normalRun()) {
                  String data = null;
                  try {
                    data = delayQueueHolder.take(queue);
                  } catch (Throwable t) {
                    normalRun(() -> log.error("延迟队列-{},获取值异常", queue, t));
                  }
                  if (StrUtil.isEmpty(data)) {
                    continue;
                  }
                  log.info("延迟队列-{},获取到值：{}", queue, data);
                  T delayMessage = JSON.parseObject(data, classType);
                  if (Objects.isNull(delayMessage)) {
                    normalRun(() -> log.error("延迟队列-{},值解析异常", queue));
                  }
                  try {
                    delayQueueExecutor.execute(delayMessage);
                  } catch (Exception e) {
                    normalRun(
                        () -> {
                          log.error("延迟队列-{} 业务处理异常", queue, e);
                          error(delayMessage);
                        });
                  }
                }
              } catch (Throwable t) {
                normalRun(
                    () -> log.error("延迟队列线程错误:{} 被异常中断结束", Thread.currentThread().getName(), t));
              }
            });
    thread.setName("DelayQueueMainThread-" + queue);
    thread.setDaemon(true);
    thread.start();
  }

  @Override
  public void destroy() {
    super.destroy();
    log.info("销毁延时队列,queueName:{}", queue);
    delayQueueHolder.destroyQueue(queue);
  }
}
